package com.toutiao.nssue

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

import java.util.Date

import com.toutiao.issue.SimpleDate
import com.toutiao.score.Topic
import com.toutiao.model.{TimeDB,TimeDict,TimeIndex}

import org.apache.log4j.Logger

import com.toutiao.score.{Scorer,Measure}
import com.toutiao.issue.UpperDict

class NssueModel(
        proc:DocProcessor,
		    //source: DocumentStore,
        source:DocStore,
		    es : IssueStore,
		    minV:Double,minS:Double,
		    locationDate: String,
		    N: Int,
		    DB_ROOT:String,
		    DB_CACHE:Int,
		    closeDate:Date
    ) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  val mVU = new UpperDict
  
  private
  val M1 = new HMap[String,Long]
  private
  val M2 = new HMap[Long,String]
  private
  var N1 = 0
  private
  var N2 = 0
  
  val Data = source
  
  val mDB = new TimeDB(DB_ROOT,DB_CACHE)
  val mTime = new TimeDict
  val K = 8
  val mIndex = new TimeIndex(K)
  //val mUpper = new UpperDict
  
  val mData = new NssueCollector(minV,mIndex,mTime,mDB,mVU)
  _log.info("Time Day Aware Issue DB")
  
  val ES = es
  
  def processOne(m:Document) = synchronized {
    if (!M1.contains(m.name)) {
      N2 += 1
      M1(m.name) = N2
      M2(N2) = m.name
      val r = mData.process(M1,M2,m,minS)
      r
    }
    else {
      ""
    }
  }
  
  def processNewOne(m:Document) = synchronized {
    if (m.vs.size == 0){
      _log.info(m.name)
    }
    if ( !M1.contains(m.name)) {
      N2 += 1
      M1(m.name) = N2
      M2(N2) = m.name
      val rs = mData.processOne(M1, M2, m, minS)
      if (rs.size > 1){
        _log.info("Merge "+rs)
      }
      if (rs.size > 0){
        rs
      }
      else {
        List("")
      }
    }
    else {
      List("")
    }
    
  }
  

  def process(ms:Array[Document]):Array[String] = {
    val ns = ms.map(e => processOne(e))
    ns
  }
  
  def processNew(ms:Array[Document]):Array[String] = {
    val ns = ms.flatMap(processNewOne)
    ns.filter(_!="").toSet.toArray
  }

  def getMaxTime(ms:Array[Document],t:Date) = {
    if (ms.size > 0){
      _getMaxTime(ms)
    }
    else {
      t
    }
  }
  
  def _getMaxTime(ms:Array[Document]) = {
    var t = ms(0).date
    for (m <- ms){
      if (m.date.getTime > t.getTime)
        t = m.date
    }
    t
  }
  
  def saveOne(n:String) {
    val m = mData.get(n)
    if (m.i != ""){
      ES.insert(m)
    }
  }
  
  def save(ns:Array[String]) {
    val ms = ns.map(mData.get(_)).filter(_.i != "")
    _log.info("Save "+ms.size)
    ES.insert(ms)
  }  
  
  def remove(T:Date) = {
    val t = SimpleDate.format(T).split(" ")(0)
    if (mDB.isIn(t)){
      val us = mDB(t).getKeys.filter(M1.contains(_))
      mData.remove(t)
      us.foreach(u => {
        val i = M1(u)
        M1 -= u
        M2 -= i
      })
      N1 += us.size
      
//      mVU.clear // Update V Upper Bound
//      mDB.keySet.foreach(t => {
//        val xs = mDB(t).getData
//        _initVU(xs)
//      })
      
      val iis = mIndex.getIndex
      for (v <- mVU.keySet) {
        val z = iis.map(index => {
          if (index.contains(v)) 1
          else 0
        }).sum
        if (z == 0)
          mVU.remove(v)
      }
        
      us.size
      
    }
    else {
      0
    }
  }
  
  private
  def _init(xs:Array[Topic]) = {
    for (x <- xs if x.m("m").size > 0 && x.m("m").size > 5) {
      val name = x.i
      N2 += 1
      M1(name) = N2
      M2(N2) = name
      val t = SimpleDate.format(x.c).split(" ")(0)
      mTime.insert(name,t)
      mIndex.insert(t,M1(name),x.v.keySet.toArray)
      //x.vs.foreach(vv => {
      //  mUpper.insert(vv._1,vv._2.toFloat)
      //})
    }
  }
  
  private
  def _initVU(xs:Array[Topic]) = {
    xs.foreach(x => {
      val vs = x.v
      vs.foreach(vv => {
         mVU.put(vv._1, vv._2.toFloat)
      })
    })
  }
  
  def init = {
    val current = SimpleDate.loadDate(locationDate)
    val T1 = SimpleDate.increment(current, -N*60)
    val T2 = SimpleDate.increment(current, 60)
    
    var ct = T1
    
    while (ct.getTime < T2.getTime){
      val t = SimpleDate.format(ct).split(" ")(0)
      _log.info(t)
      mDB.init(t)
      val xs = mDB(t).getData
      _log.info("Got "+xs.size)
      _init(xs)
      _initVU(xs)
      _log.info("Build Index Done")
      
      ct = SimpleDate.increment(ct, 60*24)
    }
    
    _log.info("VU size "+mVU.size)
    
  }  
  
  def run(m:Int) = synchronized {
    var current = SimpleDate.loadDate(locationDate)
    val rt = "create_time"
    var eT = Data.getMaxTime(rt)
    
    while (1 == 1) {
    //while (current.getTime < closeDate.getTime) {
      _log.info(SimpleDate.format(eT))
      val x = SimpleDate.increment(current, m)
      if (x.getTime <= eT.getTime) {
        _log.info("Current Date "+SimpleDate.format(current))
        _log.info("End Date "+SimpleDate.format(eT))
        val T1 = current
        val T2 = SimpleDate.increment(T1, m)
        
        val rs = Data.retrieve(T1,T2)
        _log.info("Got Raw Doc "+rs.size)
        val ms = proc.process(rs)
        _log.info("Got Doc "+ms.size)
        val mMaxTime = getMaxTime(ms,T2)
        _log.info("Process ... "+ms.size)
        val ns = processNew(ms)
        _log.info("Process Done")
        
        _log.info("Got Event "+ns.size)
        _log.info("Save ... "+ns.size)
        if (ns.size > 0)
          save(ns)
        _log.info("Save Done")
        
        val T = SimpleDate.increment(current, -60*N)
       
        _log.info(SimpleDate.format(T))
       
        val n = remove(T)
        
        _log.info("Remove News "+n)
        
        _log.info("Current Doc Size "+M1.size)
        _log.info("MAX_ID "+N2)
        _log.info("MIN_ID "+N1)
        val now = SimpleDate.current
        _log.info("mMaxTime "+SimpleDate.format(mMaxTime)+" "+SimpleDate.format(current))
        current = 
          if (mMaxTime.getTime > current.getTime) 
            if (mMaxTime.getTime <= now.getTime)
              mMaxTime
            else
              now
          else T2
          
        SimpleDate.saveDate(current,locationDate)  
      }
      else {
        val T = SimpleDate.current
        _log.info("Sleep Here "+SimpleDate.format(T))
        Thread.sleep(30*1000)
        eT = Data.getMaxTime(rt)        
      }
      
    }    
    
  }
  
}


import com.toutiao.AppProp

object NssueProcess {
  
  val _log = Logger.getLogger(this.getClass.getName)
  
  def run(prop:AppProp) {
    
    val source = new DocStore(
    				prop("ES_DOC_NAME"),
    				prop("ES_DOC_URI"),
    				prop("ES_DOC_PORT").toInt,
    				prop("ES_DOC_INDEX"),
    				prop("ES_DOC_TYPE")
    				)
    
    val es = new IssueStore(
    				prop("ES_ISSUE_NAME"),
    				prop("ES_ISSUE_URI"),
    				prop("ES_ISSUE_PORT").toInt,
    				prop("ES_ISSUE_INDEX"),
    				prop("ES_ISSUE_TYPE")
    				)
    
    val locationDate = prop("LOCATION_DATE")
    val N = prop("TIME_SIZE").toInt
    val minV = prop("MIN_V").toDouble
    val minS = prop("MIN_S").toDouble    

    val DB_ROOT = prop("DB_ROOT")
    val DB_CACHE = prop("DB_CACHE").toInt
    
    val locationDocData = prop("LOCATION_DOC_DATA")
    _log.info("Get Measure ... ")
    val measure = Measure.get(prop("LOCATION_MEASURE"))
    _log.info("Get Measure Done ")
    val DOC_NUM = prop("DOC_NUM").toInt
    val scorer = new Scorer(measure,DOC_NUM)
    
    val proc = new DocProcessor(DocData.getSource(locationDocData),scorer)
    
    val closeDate = SimpleDate.parse(prop("CLOSE_DATE"))
    
    val m = new NssueModel(proc,source,es,minV,minS,locationDate,N,DB_ROOT,DB_CACHE,closeDate)
    m.init
    val K = prop("TIME_STEP").toInt
    m.run(K)
  }
  
}



